package com.sitech.ibnms.c3p.task;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.sitech.ibnms.c3p.Application;
import com.sitech.ibnms.c3p.command.CommandParamsService;
import com.sitech.ibnms.c3p.command.CommandSupport;
import com.sitech.ibnms.c3p.command.CommandType;
import com.sitech.ibnms.c3p.domain.*;
import com.sitech.ibnms.c3p.mappers.C3pManualTaskMapper;
import com.sitech.ibnms.c3p.mappers.C3pTaskMapper;
import com.sitech.ibnms.c3p.mappers.C3pTaskNodeIOMapper;
import com.sitech.ibnms.c3p.mappers.C3pTaskNodeMapper;
import com.sitech.ibnms.c3p.service.C3pScheduleService;
import com.sitech.ibnms.c3p.service.ScriptExecutor;
import com.sitech.ibnms.c3p.task.listeners.TaskCompleteListener;
import com.sitech.ibnms.c3p.task.listeners.TaskFailedListener;
import com.sitech.ibnms.c3p.task.listeners.TaskStartListener;
import com.sitech.ibnms.c3p.task.notice.ExecutePolicy;
import com.sitech.ibnms.c3p.task.notice.NoticeHandleService;
import com.sitech.ibnms.c3p.util.C3pConstants;
import com.sitech.ibnms.c3p.util.TextDiffUtil;
import com.sitech.ismp.messageObject.cc.CccpResults;
import com.sitech.ismp.workflow.dao.ProcessInstanceDAO;
import com.sitech.ismp.workflow.dao.TaskDAO;
import com.sitech.ismp.workflow.domain.ProcessInstance;
import com.sitech.ismp.workflow.domain.Task;
import com.sitech.ismp.workflow.engine.WorkFLowConst;
import com.sitech.ismp.workflow.util.DES3;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;
import request.ScriptRequest;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.*;

import static com.sitech.ibnms.c3p.Application.sneakyThrow;
import static org.apache.commons.lang3.ObjectUtils.defaultIfNull;

/**
 * Created by liujhc on 7/9.
 * 核心逻辑 包括 启动任务 推动节点以及结束任务 失败策略
 */
@Component
public class C3pTaskFlowServiceImpl implements C3pTaskFlowServiceExtends {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private WorkflowService workflowService;
    @Autowired
    private C3pTaskMapper c3pTaskMapper;
    @Autowired
    private TaskDAO taskDAO;
    @Autowired
    private C3pTaskNodeMapper nodeMapper;
    @Autowired
    private C3pServerLogTaskService logTaskService;
    @Autowired
    private ProcessInstanceDAO processInstanceDAO;
    @Autowired
    @SuppressWarnings("MismatchedQueryAndUpdateOfCollection")
    private List<TaskStartListener> taskStartListeners;
    @Autowired
    @SuppressWarnings("MismatchedQueryAndUpdateOfCollection")
    private List<TaskCompleteListener> taskCompleteListeners;
    @Autowired
    @SuppressWarnings("MismatchedQueryAndUpdateOfCollection")
    private List<TaskFailedListener> taskFailedListeners;
    @Autowired
    private C3pTaskNodeIOMapper c3pTaskNodeIOMapper;
    @Autowired
    private C3pManualTaskMapper c3pManualTaskMapper;
    @Autowired
    private CommandParamsService paramsService;
    @Autowired
    private C3pScheduleService scheduleService;
    @Autowired
    C3pTaskNodeIOMapper ioMapper;
    @Autowired
    private NoticeHandleService noticeHandleService;
    @Autowired
    private ScriptExecutor executor;
    @Autowired
    private ObjectMapper objectMapper;

    @Override//c3ptask
    public int startTask(String c3pTaskId, String userId, String title) {
        return startTask(c3pTaskId, userId, title, Collections.<Integer, ExecutePolicy>emptyMap(), Collections.<String, String>emptyMap());
    }

    @Override
    public int startTaskWithExecutePolicies(String c3pTaskId, String userId, String title, Map<Integer, ExecutePolicy> executePolicies) {
        return startTask(c3pTaskId, userId, title, executePolicies, Collections.<String, String>emptyMap());
    }

    @Override//真实的实现类
    public int startTask(String c3pTaskId, String userId, String title, Map<String, String> context) {
        return startTask(c3pTaskId, userId, title, Collections.<Integer, ExecutePolicy>emptyMap(), context);
    }

    @Override
    public int startTask(String c3pTaskId, final String userId, final String initTitle, Map<Integer, ExecutePolicy> executePolicies, final Map<String, String> context) {
        int serialNo = 0;
        try {
            int parentSerialNo = NumberUtils.toInt(context.get(C3pConstants.CONTEXT_PARENT_SERIAL_NO), 0);
            int parentWfTaskId = NumberUtils.toInt(context.get(C3pConstants.CONTEXT_PARENT_WF_TASK_ID), 0);
            char notifyFlag = '\0'; // 用第一个不可见的字符作为控制字符，标识需要何种方式通知userId
            if (context.containsKey(C3pConstants.CONTEXT_MAIL_NOTIFY)) {
                notifyFlag |= C3pConstants.FLAG_MAIL_NOTIFY;
            }
            if (context.containsKey(C3pConstants.CONTEXT_SMS_NOTIFY)) {
                notifyFlag |= C3pConstants.FLAG_SMS_NOTIFY;
            }
            final String title = notifyFlag + initTitle;
            String scheduleId = StringUtils.defaultString(context.get(C3pConstants.CONTEXT_SCHEDULE_ID));
            logger.info("startTask: c3pTaskId={}, userId={}, title={}, parentSerialNo={}, parentWfTaskId={}, scheduleId={}",
                    c3pTaskId, userId, title, parentSerialNo, parentWfTaskId, scheduleId);
            // 由c3pTaskId获取processId
            final C3pTask c3pTask = c3pTaskMapper.findById(c3pTaskId);
            if (c3pTask == null) {
                if (scheduleId != null) {
                    logger.error("删除错误调度数据[scheduledId={}, 无对应c3pTaskId={}]", scheduleId, c3pTaskId);
                    scheduleService.unscheduleTask(scheduleId);
                    throw new IllegalArgumentException("找不到对应的任务(c3pTaskId=" + c3pTaskId + ")，已移除此调度(scheduleId=" + scheduleId + ")");
                }
                throw new IllegalArgumentException("找不到对应的任务(c3pTaskId=" + c3pTaskId + ")");
            }
            int processId = c3pTask.getProcessId();
            Assert.state(processId > 0, String.format("C3pTask[c3pTaskId=%s] has no processId!", c3pTaskId));
            logger.info("processId is {}", processId);

            logger.info("Invoke workflowService.startProcess: processId={}, userId={}, parentSerialNo={}, parentWfTaskId={}, title={}",
                    processId, userId, parentSerialNo, parentWfTaskId, title);
            //调用了服务管理的api，服务管理的task
            //scheduleId该字段是废物利用，填上我们需要存的值
            //parentSerialNo 父流程实例号
            //parentWfTaskId 父流程环节实例号
            Task task = workflowService.startProcess(processId, userId, title, null, null, scheduleId, parentSerialNo, parentWfTaskId);
            serialNo = task.getSerialNo();//流水号,流程实例号
            //服务管理 dao查询配置流程信息
            //process标识流程配置数据 ，一个流程只有一个，唯一标识一个流程， 代表类似于java中类的概念,
            //ProcessInstance流程实例对象，代表类似于java中类的对象的概念
            //环节 activity ，类似于类里面的方法 ，不是实例  是流程里环节的配置
            //task 服务管理 环节的实例  注意：与集中控制的task概念不一样。
            //服务管理 流程 与环节   集中控制  任务  指令 ,
            //
            final ProcessInstance processInstance = processInstanceDAO.queryProcessInstanceInfoBySerialNo(serialNo);
            // 执行监听器
            //用于管理 任务执行相关事件 用于扩展功能
            for (TaskStartListener taskListener : taskStartListeners) {
                try {
                    taskListener.onStart(c3pTask, processInstance, context);
                } catch (Throwable e) {
                    logger.error(e.getMessage(), e);
                }
            }
            //写到日志表 给页面展示 用
            logTaskService.writeTaskLog(task, (parentSerialNo != 0 ? "子任务" : "任务") + "【${taskName}】开始启动，执行ID为" + serialNo + " ...");

            // 保存执行策略参数
            //
            storeExecutePolicy(task, executePolicies);

            // 提交开始环节
            //与分支 前后节点 都会拿出
            final List<Task> tasks = workflowService.submitTask(task.getTaskId(), userId, title, null, serialNo);
            // 异步执行非流程实例创建逻辑，以便快速返回serialNo
            new Thread(new Runnable() {
                @Override
                public void run() {
                    // 启动处理业务节点...
                    for (Task nextTask : tasks) {
                        self().startNode(nextTask.getTaskId(), userId, title);
                    }
                }
            }).start();
            return serialNo;
        } catch (Throwable t) {
            if (serialNo != 0) {
                logger.error(String.format("TaskTerminated: startTask Error[serialNo=%s]", serialNo), t);
                terminate(serialNo, userId, t);
            }
            throw sneakyThrow(t);
        }
    }

    @Async
    @Override
    public void startNode(int taskId, String userId, String title) {
        startNode(taskId, userId, title, false);
    }
//执行 集中控制 业务节点对应的指令
    private void startNode(int taskId, String userId, String title, boolean isManual) {
        Task task = null;
        C3pTaskNode node = null;
        int serialNo = 0;
        try {
            task = taskDAO.queryTaskByTaskId(taskId);
            serialNo = task.getSerialNo();
            logger.info("startNode: serialNo={}, activityId={}, taskId={}, userId={}, title={}",
                    serialNo, task.getCurrActId(), taskId, userId, title);
            //task.getCurrActId() 当前环节id
            //node ：C3pTaskNode
            node = nodeMapper.queryByProcessIdAndActivityId(task.getProcessId(), task.getCurrActId());
            logTaskService.writeTaskNodeLog(task, "环节【${nodeName}】开始启动(" + taskId + ") ...");
            logger.info("{}", node);
            //发短信 发邮件 发通知的
            noticeHandleService.sendStartNodeNotice(node, serialNo);

            if (!isManual) {//判断是否为手动执行
                //手动执行isManual==true

                //根据入参中的配置策略进行下一步操作
                ExecutePolicy executePolicy = paramsService.getExecutePolicy(task);
                logger.debug("task[serialNo={}, activityId={}, taskId={}] ExecutePolicy is {}", serialNo, task.getCurrActId(), taskId, executePolicy);
                switch (executePolicy) {
                    case SKIP:
                        //告诉服务流程 开始走下一步 bb并打印日志
                        submitNode(task.getTaskId(), task.getCurrDealer(), task.getTitle());
                        logTaskService.writeTaskNodeLog(task, "环节【${nodeName}】执行策略为【" + executePolicy + "】，跳过本环节执行");
                        return;
                    case PAUSE:
                        C3pManualTask manualTask = new C3pManualTask(task);
                        manualTask.setC3pTaskId(node.getC3pTaskId());
                        manualTask.setTaskFrom(C3pConstants.MANUAL_TASK_FROM_EXECUTE);
                        c3pManualTaskMapper.addManualTask(manualTask);//插入到人工代办表 需要人工推
                        logTaskService.writeTaskNodeLog(task, "环节【${nodeName}】执行策略为【" + executePolicy + "】，等待人工处理");
                        return;
                }
            }
            switch (node.getActionType()) {//得到当前为指令还是子任务
                case COMMAND:// 指令
                    //执行
                    CommandSupport.newCommand(node.getActionId()).execute(node.getC3pTaskId(), task);
                    break;
                case TASK: // 子任务
                    Map<String, String> context = new HashMap<String, String>();
                    context.put(C3pConstants.CONTEXT_PARENT_SERIAL_NO, String.valueOf(serialNo));
                    context.put(C3pConstants.CONTEXT_PARENT_WF_TASK_ID, String.valueOf(task.getTaskId()));
                    context.put(C3pConstants.CONTEXT_SCHEDULE_ID, processInstanceDAO.queryProcessInstanceInfoBySerialNo(serialNo).getWf_num());
                    self().startTask(node.getActionId(), userId, title, Collections.<Integer, ExecutePolicy>emptyMap(), context);
                    break;
            }
        } catch (Throwable t) {
            logger.error(String.format("TaskTerminated: startNode Error[serialNo=%s]", serialNo), t);
            Assert.notNull(task, String.format("taskDAO.queryTaskByTaskId(%s)==null", taskId));
            Assert.notNull(node, String.format("nodeMapper.queryByProcessIdAndActivityId(%s, %s)==null", task.getProcessId(), task.getCurrActId()));
            //进行失败策略
            failPolicy(task, t);
            //terminate(serialNo, userId, t);
            logTaskService.writeTaskNodeLog(task, "环节【${nodeName}】启动异常：" + t.getMessage());
        } finally {
            if (task != null && node != null) {//业务逻辑允许的失败
                // 已经被C3pNodeWatchdogJob设置为跳过处理
                //任务是否被标记为 跳过
                boolean isDealSkipped = "-1".equals(task.getDealType());
                // 环节配置了超时时限
                boolean isTimeLimitSet = node.getTimeLimitSeconds() > 0;
                //如果被标记为跳过或超时
                if (isDealSkipped || isTimeLimitSet) {
                    if (isDealSkipped) {
                        task.setDealType("");
                    }
                    //什么时候到什么时间点 超时 通过DateAllow 检查超时
                    if (isTimeLimitSet) { // 如果环节设置了执行时限，就将截止时间更新到Task表中，供超时检查任务处理
                        task.setDateAllow(DateUtils.addSeconds(new Date(), node.getTimeLimitSeconds()));
                    }
                    //大写DAO为服务管理dao
                    //集中控制 相关的表 用mapper
                    taskDAO.updatTask(task);
                }
            }
        }
    }

    @Override
    //该节点Node为集中控制里的概念 通过该方法 执行我们处理逻辑之后的操作 ，告诉服务管理 我们执行完毕，推动
    //我们的业务逻辑已经执行完  ，流程向下走会调用此方法  返回接下来 需要执行的环节
    public void submitNode(final int taskId, final String userId, final String title) {
        Task task = null;
        C3pTaskNode node = null;
        List<Task> tasks = Collections.emptyList();
        int serialNo = 0;
        try {
            task = c3pManualTaskMapper.queryTaskInfoById(taskId);
            Assert.notNull(task, "环节已经结束");
            node = nodeMapper.queryByProcessIdAndActivityId(task.getProcessId(), task.getCurrActId());
            serialNo = task.getSerialNo();
            logger.info("submitNode: serialNo={}, activityId={}, taskId={}, userId={}, title={}",
                    serialNo, task.getCurrActId(), taskId, userId, title);
            //推动 服务管理 流程走 返回接下来（下一步） 需要执行的环节
            tasks = workflowService.submitTask(task.getTaskId(), userId, title, null, serialNo);
            task.setCurrDealer(userId); // 设置新的用户Id，用于写日志
            logTaskService.writeTaskNodeLog(task, "环节【${nodeName}】执行完毕");
            noticeHandleService.sendEndNodeNotice(node, serialNo, true);
        } catch (Exception ex) {
            try {
                if (task == null) {
                    task = c3pManualTaskMapper.queryHaveDoneTaskInfoById(taskId);
                    node = nodeMapper.queryByProcessIdAndActivityId(task.getProcessId(), task.getCurrActId());
                }
                logTaskService.writeTaskNodeLog(task, "环节【${nodeName}】执行异常：" + ex.getMessage());
                noticeHandleService.sendEndNodeNotice(node, serialNo, false);
            } finally {
                logger.error(String.format("TaskTerminated: submitNode Error[serialNo=%s]", serialNo), ex);
                terminate(serialNo, userId, ex);
            }
        } finally {
            c3pManualTaskMapper.deleteByTaskId(taskId);
        }
        //可能每一步 都返回一个或多个环节（例如 与分支 ）
        for (Task nextTask : tasks) {
            //判断是否为结束环节
            //如果非结束环节
            if (nextTask.getActivityType() != WorkFLowConst.ACTIVITY_TYPE_COMPLETION) {
                self().startNode(nextTask.getTaskId(), userId, title);
            } else {//如果是结束环节
                ProcessInstance processInstance = processInstanceDAO.queryProcessInstanceInfoBySerialNo(serialNo);
                boolean subInstance = processInstance.getParentTaskId() != 0;
                try {
                    // 监听器
                    C3pTask c3pTask = c3pTaskMapper.queryBySerialNo(serialNo);
                    if (c3pTask == null) {
                        logger.warn("Cannot find C3pTask by serialNo {}", serialNo);
                    }
                    for (TaskCompleteListener taskListener : taskCompleteListeners) {
                        try {
                            taskListener.onComplete(c3pTask, processInstance);
                        } catch (Throwable e) {
                            logger.error(e.getMessage(), e);
                        }
                    }

                    if (subInstance) {//判断是否为子时例
                        logTaskService.writeTaskLog(task, "子任务【${taskName}】执行完毕");
                    } else try {
                        logTaskService.writeTaskLog(task, "任务【${taskName}】执行完毕");
                        logTaskService.writeTaskLog(task, C3pConstants.FLAG_TASK_COMPLETE);
                        cleanUp(serialNo);//结束流程
                    } catch (Exception e) {
                        //noinspection ThrowableResultOfMethodCallIgnored
                        logTaskService.writeTaskLog(task, defaultIfNull(e.getCause(), e).getMessage());
                    }
                    logger.info("TaskCompleted: processId={}, serialNo={}, userId={}, title={}", task.getProcessId(), serialNo, userId, title);
                } finally {
                    // 如果当前流程是子任务触发，需要提交父环节
                    if (subInstance) {
                        logger.info("submitParentNode: taskId={}, uesrId='{}', title='{}'",
                                processInstance.getParentTaskId(), userId, title);
                        submitNode(processInstance.getParentTaskId(), userId, title);//告诉父任务 执行完毕 可以 继续执行。
                    }
                }
            }
        }
    }

    @Transactional
    public void cleanUp(int rootSerialNo) {
        //流程实例的C3P_TASK_NODE_IO表里的数据复制到c3ptasknodeiohis
        ioMapper.insertToHisByRootSerialNo(rootSerialNo);
        //根据 SerialNo将C3P_TASK_NODE_IO表里复制过的数据删掉
        ioMapper.deleteByRootSerialNo(rootSerialNo);
    }


    /**
     * 失败策略，通过给定的配置来进行不同的操作
     *
     * @param task
     * @param throwable
     */
    @Override
    public void failPolicy(Task task, Throwable throwable) {
        logger.warn("task[serialNo={}, activityId={}, taskId={}]到达失败策略: {}", task.getSerialNo(), task.getCurrActId(), task.getTaskId(), throwable.getMessage());
        c3pManualTaskMapper.deleteByTaskId(task.getTaskId());
        C3pTaskNode node = nodeMapper.queryByProcessIdAndActivityId(task.getProcessId(), task.getCurrActId());
        noticeHandleService.sendEndNodeNotice(node, task.getSerialNo(), false);
        // 按配置的失败策略处理异常
        FailPolicy failPolicy = defaultIfNull(node.getFailPolicy(), FailPolicy.DEFAULT);
        if (failPolicy.equals(FailPolicy.DEFAULT)) {// 手动执行流程图任务时, 默认失败策略为人工处理
            C3pTask c3pTask = c3pTaskMapper.findById(node.getC3pTaskId());
            if (TaskConfType.topology.equals(c3pTask.getTaskConfType())) {
                ProcessInstance processInstance = processInstanceDAO.queryProcessInstanceInfoBySerialNo(task.getSerialNo());
                //判断是否是手动执行任务
                if (processInstance != null && StringUtils.isBlank(processInstance.getWf_num())) {
                    logTaskService.writeTaskNodeLog(task, "环节【${nodeName}】未设置失败策略，当前任务为人工触发，等待人工处理");
                    failPolicy = FailPolicy.PAUSE;
                }
            }
        }
        switch (failPolicy) {
            case CONTINUE:
                logTaskService.writeTaskNodeLog(task, "环节【${nodeName}】失败策略【" + FailPolicy.CONTINUE.name() + "】,跳过错误，继续执行任务");
                submitNode(task.getTaskId(), task.getCurrDealer(), task.getTitle());
                break;
            case PAUSE:
                C3pManualTask manualTask = new C3pManualTask(task);
                manualTask.setC3pTaskId(node.getC3pTaskId());
                manualTask.setTaskFrom(C3pConstants.MANUAL_TASK_FROM_FAIL);
                c3pManualTaskMapper.addManualTask(manualTask);
                logTaskService.writeTaskNodeLog(task, "环节【${nodeName}】失败策略【" + FailPolicy.PAUSE.name() + "】,暂停任务，等待人工处理");
                break;
            case BREAK:
                logTaskService.writeTaskNodeLog(task, "环节【${nodeName}】失败策略【" + FailPolicy.BREAK.name() + "】,结束任务");
                terminate(task.getSerialNo(), task.getCurrDealer(), throwable);
                break;
            default:
                logTaskService.writeTaskNodeLog(task, "环节【${nodeName}】未设置失败策略，终止流程");
                terminate(task.getSerialNo(), task.getCurrDealer(), throwable);
        }
    }

    @Override
    public void terminate(int serialNo, String userId, String title) {
        terminate(serialNo, userId, new Exception(title));
    }

    public void terminate(int serialNo, String userId, Throwable throwable) {
        StringWriter stringWriter = new StringWriter();
        throwable.printStackTrace(new PrintWriter(stringWriter));
        logger.info("terminate process instance: serialNo={}, userId={}, throwable={}", serialNo, userId, stringWriter.toString());
        ProcessInstance processInstance = processInstanceDAO.queryProcessInstanceBySerialNo(serialNo);
        if (processInstance == null) {
            logger.warn("试图终止任务【serialNo={}】，但是任务已经终止！", serialNo);
            return;
        }
        String title = throwable.getMessage();
        try {
            workflowService.terminate(serialNo, userId, throwable.getMessage());
        } catch (Throwable t) {
            logger.error(t.getMessage(), t);
        } finally {
            c3pManualTaskMapper.deleteByExecId(serialNo);
            boolean subInstance = processInstance.getParentSerialNo() != 0;
            try {
                C3pTask c3pTask = c3pTaskMapper.queryBySerialNo(serialNo);
                if (c3pTask == null) {
                    logger.warn("Cannot find C3pTask by serialNo %s", serialNo);
                } else {
                    // 执行任务失败监听器
                    for (TaskFailedListener taskListener : taskFailedListeners) {
                        try {
                            taskListener.onFailed(c3pTask, processInstance, title);
                        } catch (Throwable e) {
                            logger.error(e.getMessage(), e);
                        }
                    }

                    Task task = new Task();
                    task.setCurrDealer(userId);
                    task.setSerialNo(serialNo);
                    if (subInstance) {
                        logTaskService.writeTaskLog(task, "子任务【${taskName}】执行异常终止：" + title);
                    } else {
                        logTaskService.writeTaskLog(task, "任务【${taskName}】执行异常终止：" + title);
                        logTaskService.writeTaskLog(task, C3pConstants.FLAG_TASK_TERMINATE);
                        cleanUp(serialNo);
                    }
                }
            } finally {
                // 如果当前流程是子任务触发，需要终止父流程
                if (subInstance) {
                    logger.info("terminateParentInstance: serialNo={}, uesrId='{}', title='{}'",
                            processInstance.getParentSerialNo(), userId, title);
                    Task parentTask = taskDAO.queryTaskByTaskId(processInstance.getParentTaskId());
                    if (parentTask != null) {
                        failPolicy(parentTask, throwable);
                    } else {
                        terminate(processInstance.getParentSerialNo(), userId, title);
                    }
                }
            }
        }
    }

    @Async
    @Override
    public void startNodeWithExecutePolicies(int taskId, String userId, String title, Map<Integer, ExecutePolicy> executePolicies) {
        // 保存执行策略
        storeExecutePolicy(taskDAO.queryTaskByTaskId(taskId), executePolicies);

        // 真正启动环节
        startNode(taskId, userId, title, true);

        // 删除人工执行的记录
        c3pManualTaskMapper.deleteByTaskId(taskId);
    }


    private void storeExecutePolicy(Task task, Map<Integer, ExecutePolicy> executePolicies) {
        Assert.notNull(task, "the param Task must not null!");
        int serialNo = task.getSerialNo();

        if (task.getPreActId() != 0) { // 在流程开始时不需要清理
            logger.info("根据serialNo删除还未执行的环节的执行策略参数,serialNo={}", serialNo);
            c3pTaskNodeIOMapper.deletePolicyParamNotExecute(serialNo);
        }

        if (executePolicies != null && !executePolicies.isEmpty()) {
            Set<Integer> activityIdSet = executePolicies.keySet();
            logger.info("根据配置的环节执行策略，添加输入参数,参数数量:{}", activityIdSet.size());
            C3pTask c3pTask = c3pTaskMapper.queryBySerialNo(serialNo);
            String c3pTaskId = c3pTask == null ? "" : c3pTask.getTaskId();
            ArrayList<C3pTaskNodeIo> nodeIos = new ArrayList<C3pTaskNodeIo>();
            for (Integer activityId : activityIdSet) {
                //如果已经过了这个环节了  那么这个环节的执行策略就不能修改了
                int paramCount = c3pTaskNodeIOMapper.selectParamByActivityNotExecutePolicy(serialNo, activityId);
                if (paramCount > 0) {
                    logger.info("环节已经执行,不能在修改执行策略,activityId={}", activityId);
                } else {
                    C3pTaskNodeIo c3pTaskNodeIo = new C3pTaskNodeIo();
                    c3pTaskNodeIo.setParamName(C3pConstants.PARAM_EXECUTE_POLICY);
                    c3pTaskNodeIo.setActivityId(activityId);
                    c3pTaskNodeIo.setParamType("IN");
                    ExecutePolicy paramValue = executePolicies.get(activityId);
                    c3pTaskNodeIo.setParamValue(paramValue != null ? paramValue.name() : "");
                    c3pTaskNodeIo.setSerialNo(serialNo);
                    c3pTaskNodeIo.setC3pTaskId(c3pTaskId);
                    c3pTaskNodeIo.setProcessId(task.getProcessId());
                    c3pTaskNodeIo.setDbTime(new Date());
                    c3pTaskNodeIo.setTaskId(0);
                    nodeIos.add(c3pTaskNodeIo);
                }
            }
            if (!nodeIos.isEmpty()) c3pTaskNodeIOMapper.insertParam(nodeIos);
        }
    }
    @Override
    public Map<String, String> executeShell(Map<String, String> params,Map<String,String> inputVars, Map<String,String> outputVars){
        Map<String, String> result=new HashMap<String, String>();
        try{
            logger.info("*********executeShell"+params.values()+",keys="+params.keySet());
            String targetIpAddress = params.get("IP");
            String targetUnitId = params.get("UNIT_ID");
            String targetUserName = params.get("USER_NAME");
            String passwd=params.get("PASSWORD");
            String port=params.get("PORT");
            int displayMode=TextDiffUtil.returnInt(params.get("DISPLAY_MODE"));
            String targetPassword = DES3.decrypt(passwd);
            inputVars.put("IP", targetIpAddress);
            inputVars.put("UNIT_ID", targetUnitId);
            inputVars.put("USERNAME", targetUserName);
            inputVars.put("PASSWORD", targetPassword);
            inputVars.put("PORT", port);
            inputVars.put("_TARGET_IP_ADDRESS", targetIpAddress);
            inputVars.put("_TARGET_UNIT_ID", targetUnitId);
            inputVars.put("_TARGET_USERNAME", targetUserName);
            inputVars.put("_TARGET_PASSWORD", targetPassword);

            ScriptRequest scriptRequest = new ScriptRequest();
            scriptRequest.setScriptType(CommandType.SHELL.getScriptType());
            scriptRequest.setContent(params.get("SCRIPT"));
            scriptRequest.setInputVars(objectMapper.writeValueAsString(inputVars));
            ObjectNode objectNode = objectMapper.createObjectNode();
            for (Map.Entry entry : outputVars.entrySet()) {
                objectNode.put(entry.getKey().toString(), "");
            }
            scriptRequest.setOutputVars(objectNode.toString());
            scriptRequest.setRequestId("shell-" + params.hashCode());
            ResourceAccount resourceAccount = new ResourceAccount();
            resourceAccount.setIpAddr(targetIpAddress);
            resourceAccount.setUserName(targetUserName);
            resourceAccount.setPassword(passwd);
            resourceAccount.setUnitId(targetUnitId);
            resourceAccount.setPort(port==null||port.equals("null")?0:Integer.parseInt(port));
            resourceAccount.setDisplayMode(displayMode);
            CccpResults cccpResults = executor.executeScript(scriptRequest, resourceAccount, Collections.<String, String>emptyMap());
            result.put("SCRIPT_CONTENT",cccpResults.getVariables().get(C3pConstants.SCRIPT_CONTENT));
            if(cccpResults.getExecFlag()!=-1){
                result.put("RUN_RESULT_STATUS","SUCCESS");
                if(displayMode==2){
                    result.put("RUN_RESULT", cccpResults.getVariables().get(C3pConstants.OUTVAR_SHELL_RESULT_CONTENT));
                }else{
                    result.put("RUN_RESULT", cccpResults.getExecResult());
                }
            }else{
                result.put("RUN_RESULT_STATUS","FAIL");
                result.put("RUN_RESULT",cccpResults.getFailedReason());
            }
            return result;
        }catch (Exception e){
            result.put("RUN_RESULT_STATUS","FAIL");
            result.put("RUN_RESULT",e.getMessage());
            return result;
        }
    }
    public Map<String, String> executeSql(Map<String, String> params,Map<String,String> inputVars){
        Map<String, String> result=new HashMap<String, String>();
        try{
            String unit_id = params.get("UNIT_ID");
            String dbDriver = params.get("DB_DRIVER");
            String dbUrl = params.get("DB_URL");
            String userName = params.get("USER_NAME");
            String passwd = params.get("PASSWORD");
            String targetPassword = DES3.decrypt(passwd);
            String ip = params.get("IP");

            inputVars.put("_TARGET_IP_ADDRESS", ip);
            inputVars.put("_TARGET_UNIT_ID", unit_id);
            inputVars.put("_TARGET_USERNAME", userName);
            inputVars.put("_TARGET_PASSWORD", targetPassword);

            ScriptRequest scriptRequest = new ScriptRequest();
            scriptRequest.setScriptType(CommandType.SQL.getScriptType());
            scriptRequest.setContent(params.get("SCRIPT"));
            scriptRequest.setInputVars(objectMapper.writeValueAsString(inputVars));
            scriptRequest.setOutputVars("{}");
            scriptRequest.setRequestId("sql-" + params.hashCode());
            ResourceAccount resourceAccount = new ResourceAccount();
            resourceAccount.setDbDriver(dbDriver);
            resourceAccount.setDbUrl(dbUrl);
            resourceAccount.setIpAddr(ip);
            resourceAccount.setPassword(passwd);
            resourceAccount.setUnitId(unit_id);
            resourceAccount.setUserName(userName);

            resourceAccount.setPassword(passwd);
            CccpResults cccpResults = executor.executeSql(scriptRequest, resourceAccount, Collections.<String, String>emptyMap());

            result.put("RUN_RESULT_STATUS",cccpResults.getExecFlag()!=-1?"SUCCESS":"FAIL");
            result.put("SCRIPT_CONTENT",cccpResults.getVariables().get(C3pConstants.SCRIPT_CONTENT));
            result.put("RUN_RESULT",cccpResults.getExecFlag()!=-1?cccpResults.getExecResult():cccpResults.getFailedReason());
            return result;
        }catch (Exception e){
            result.put("RUN_RESULT_STATUS","FAIL");
            result.put("RUN_RESULT",e.getMessage());
            return result;
        }
    }
    // 从spring容器中获取实例，以使同类中的@Async异步方法执行生效
    private C3pTaskFlowServiceExtends self() {
        if (self == null) {
            return self = Application.getBean(C3pTaskFlowServiceExtends.class);
        }
        return self;
    }

    private C3pTaskFlowServiceExtends self;
}
